package org.example.rocketmqstudy.producer;

import io.netty.channel.DefaultChannelId;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

public class OrderlySuspendProducer {
    public static AtomicLong idGenerator = new AtomicLong(1000);

    public static void main(String[] args) throws InterruptedException, MQClientException, MQBrokerException, RemotingException {
        DefaultChannelId.newInstance(); // 解决 RemotingTimeoutException: invokeSync call timeout 异常

        DefaultMQProducer producer = new DefaultMQProducer("orderly-producer");
        producer.setNamesrvAddr("8.138.36.56:9876");
        producer.start();

        for (int i = 0; i < 20; i++) {
            Message message = new Message("test-orderly-suspend", ("orderly-message-" + i).getBytes());
            producer.send(message, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    Integer id = (Integer) arg;
                    int index = id % mqs.size();
                    return mqs.get(index);
                }
            }, 1);
            String now = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());
            System.out.printf("%s [%s] 发送消息，msg：%s %n", now, Thread.currentThread().getName(), new String(message.getBody()));
        }

        producer.shutdown();
    }
}
